package com.servlet;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.beans.UserInfo;
import com.beans.WordCountInfo;
import com.constant.Constant;

import net.sf.json.JSONArray;


@WebServlet("/MR_WordCountServlet")
public class MR_WordCountServlet extends HttpServlet {
	//private static final long serialVersionUID = 1L;
	protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		//System.setProperty("HADOOP_USER_NAME","root");
		try {
			
			// 创建一个作业
			Job job = Job.getInstance();
	
			// 指定main函数所在的类
			//job.setJarByClass(WordCountTest.class);
	
			// 设定map 相关的配置
			job.setMapperClass(WordCountMapper.class);
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(LongWritable.class);
	
	
			String filePath=request.getParameter("filePath");
			FileInputFormat.setInputPaths(job, new Path(Constant.HDFS_PATH+filePath));
	
			
			// 设定 reduce 相关的配置
			job.setReducerClass(WordCountMyReducer.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(LongWritable.class);
	
			// 因为如果目标目录存在,将出错,所以可以先将目标删除
			URI uri = new URI(Constant.HDFS_PATH);
			//Configuration conf = new Configuration();
			FileSystem fs = FileSystem.get(uri, Constant.CONF);
			
			UserInfo user=(UserInfo)request.getSession().getAttribute("session_user");
			String userRoot = user.getUserName();
			fs.delete(new Path("/"+userRoot+"/workcounttmp"), true);
	
			// 指明计算完成以后,输出结果放在哪里
			FileOutputFormat.setOutputPath(job, new Path(Constant.HDFS_PATH+userRoot+"/workcounttmp"));
			
			// 提交作业
			job.waitForCompletion(true); // true 表示在执行作业的时候输出提示信
			
			System.out.println("作业完成");
			
			//读出作业结果
			Path path=new Path(Constant.HDFS_PATH+userRoot+"/workcounttmp/part-r-00000");
			FSDataInputStream fsInput = fs.open(path);
			
			BufferedReader br = new BufferedReader (new InputStreamReader(fsInput,"utf-8"));
			List<WordCountInfo> wordCountList = new ArrayList<>();
			
			String str = null;
			while((str = br.readLine())!=null) {
				System.out.println(str);
				String data[] = str.split("\t");
				String word = data[0];
				Integer count = 0;
				try {
					count=Integer.parseInt(data[1]);
				}
				catch(Exception e){
				}
				WordCountInfo info = new WordCountInfo(count,word);
				wordCountList.add(info);
			}
			response.setContentType("text/html;charset=utf-8");
			
			//我们要把wordCountList转成json格式，传给前台
			JSONArray jsonObj = JSONArray.fromObject(wordCountList);
			
			System.out.println(jsonObj.toString());
			
			response.getWriter().println(jsonObj);
		}
		catch(Exception ex){
			ex.printStackTrace();
		}
	}
	
	
	// KEYIN LongWritable key 代表行号
	// VALUEIN Text value 代表当前进行处理的那行数据
	// KEYOUT 往后面写的key的类型
	// VALUEOUT 往后面写的value的类型
	static class WordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			String[] wordList = line.split(" ");

			for (String word : wordList) {
				context.write(new Text(word), new LongWritable(1));
			}
		}
	}

	static class WordCountMyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
		LongWritable n = new LongWritable();
		long count = 0;

		protected void reduce(Text key, Iterable<LongWritable> values, Context context)
				throws IOException, InterruptedException {

			for (LongWritable i : values) {
				count += i.get();
			}

			n.set(count);
			context.write(key, n);
			count=0;
		}
	}
}
